package spark.sql

import java.util.Properties

import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by ibf on 2018/2/4.
 */
/**
 * 1、将hive中表数据加载到mysql中
 * 2、将hive表和mysql做join
 * 3、将join之后的数据按照parquet格式写入HDFS
 *
 * idea要连接mysql，给定参数（元数据）
 *    根据hive.metastore.uris参数，采用不同策略
 *  1、如果没有给定具体的值，需要将hive元数据信息和相关的jdbc配置信息添加到sparkconfigration上
 *      并将元数据库的驱动包加载到当前项目的classpath中
 *  2、给定具体的参数，通过metastore服务连接上元数据信息
 */
object HiveJoinMysqlDataSparkSql {
  def main(args: Array[String]) {
    //0、构建常量
    val driver = "com.mysql.jdbc.Driver"
    val url = "jdbc:mysql://bigdata-01:3306/test"
    val password = "123456"
    val user = "root"
    val props = new Properties()
    props.put("user",user)
    props.put("password",password)

    //1、创建spark上下文
    val conf = new SparkConf()
        .setAppName("HiveJoinMysqlDataSparkSql")
        .setMaster("local[*]")
    val sc = SparkContext.getOrCreate(conf)
    //这里要考虑是否要读取hive的数据，或者使用HQL
    //如果需要：就使用HiveContext，如果不需要，就使用SQLContext
    val sqlContext = new HiveContext(sc)


    /**
     * 需求一：将dept表，加载到mysql当中
     *
     *    通过read/write的API将数据同步到mysql中
     *    1、先考虑如何读取hive中的表数据
     *          （PermSize内存溢出：添加VM  -XX:PermSize=128M -XX:MaxPermSize=256M）
     *    2、将读取到的hive表数据（DataFrame）写出到Mysql中
     */
     sqlContext.read
          .table("class19.dept")
          .write
       //写出去的时候，指定模式
       /**
        * spark的writeAPI写出数据的时候：
        *     Append   追加
        *     Overwrite 覆盖（将原表删除，重新创建，导入数据）
        *     ErrorIfExists 表存在就报错
        *     Ignore   表存在就不做任何操作
        *   没有实现insert or update的操作（基于数据记录进行更新或者插入的操作）
        *   如果想要实现这样的功能，只能自己使用jdbc来实现
        */
          .mode(SaveMode.Overwrite)
          .jdbc(url,"hive_dept",props)

    /**
     * 需求二：将hive表（emp）和mysql中的表做join操作
     *    1、读取mysql中的数据，形成DataFrame
     *    2、将DataFrame注册成临时表（tmp_mysql_dept）
     *    3、写sql对两张表信息做join
     */

        sqlContext.read
                //方式一：如果通过这个方式加载数据的话，数据只会全部都在1个分区中
                //会影响spark任务的并行度
                //.jdbc(url,"hive_dept",props)
          /**
           * 方式二：
           *  url: String,
              table: String,
             columnName: String, // 给定分区字段的列名称
               lowerBound: Long, // 给定计算范围下界
             upperBound: Long, // 给定计算范围上界
             numPartitions: Int, // 给定分区个数
              connectionProperties: Properties)
           步长：（upperBound-lowerBound）/numPartitions
           第一个区间：负无穷到（lowerBound+步长）
           第二个区间：（lowerBound+步长）到（lowerBound+步长+步长）
           ....
           最后一个区间：上一个区间的最大值到正无穷

           */
                  //.jdbc(url,"hive_dept","deptno",20,30,3,props)
            //方式三：手动给定每个分区的范围
                    .jdbc(url,"hive_dept",Array[String]("deptno < 25",
                 "deptno >= 25 AND deptno < 28",
                 "deptno >= 28"),props)
              //注册成临时表
              .registerTempTable("tmp_mysql_dept")

    val joinResultDF = sqlContext.sql(
      """
        |SELECT a.*,b.dname,b.loc
        |FROM
        |class19.emp a
        |join
        |tmp_mysql_dept b
        |on
        |a.deptno = b.deptno
      """.stripMargin)
    joinResultDF.cache()

    joinResultDF.show

    /**
     * 需求三：将数据保存为Parquet格式，存入HDFS之上
     *
     * 保存的时候遇到的报错：
     *  Reference 'deptno' is ambiguous, could be: deptno#13, deptno#3.;
     *  字段混淆，因为这张表中有两个字段是相同名称的
     */
    joinResultDF.write
          .mode(SaveMode.Overwrite)
          //.partitionBy("deptno")
          .format("parquet")
          .save("hdfs://bigdata-01:8020/spark/result/parquet")


    //线程等待
    Thread.sleep(1000000l)
  }

}
